package com.funny.rabbitmq.pubsub.direct;

import com.funny.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class Consumer {

    private final static String EXCHANGE_NAME = "logs.direct";
    private final static String QUEUE_NAME = "console";
    private final static String ROUTING_KEY = "";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "waring");

        /**
         * 消费者消费参数
         * 1.消费队列名称
         * 2.是否自动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */
        channel.basicConsume(QUEUE_NAME, false, (consumerTag, message) -> {
            String msg = new String(message.getBody());
            log.info("consumerTag={}, message={}", consumerTag, msg);
            boolean result = handleMsg(msg);
            long deliveryTag = message.getEnvelope().getDeliveryTag();
            String routingKey = message.getEnvelope().getRoutingKey();
            log.info("deliveryTag={}, routingKey={}", deliveryTag, routingKey);
            if (result) {
                channel.basicAck(deliveryTag, false);
            } else {
                channel.basicNack(deliveryTag, false, true);
//                channel.basicReject(deliveryTag, false);
            }
        }, consumerTag -> {
            log.info("consumerTag={}", consumerTag);
        });
        System.out.println("等待消费...");
    }

    private static boolean handleMsg(String msg) {
        String[] strs = msg.split("_");
        int n = Integer.parseInt(strs[1]);
        return n > 2;
    }
}
